package org.fuxia.im.core.server;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;

import org.fuxia.im.core.entries.ECPacket;
import org.fuxia.im.core.entries.ECSession;
import org.fuxia.im.core.socket.impl.ECNioServer;
/**
 * 服务器Session管理器
 *
 */
public class ECSessionMananger
{	
	//离线消息队列，每个用户拥有一个离校消息队列
	ConcurrentHashMap<String, ConcurrentLinkedQueue<ECPacket>>      offlineMessage  = null ;
	//存放临时session
	ConcurrentMap<String, ECSession>		allSessions			= null;
	//存放已经登录成功的session
	ConcurrentMap<String, ECSession>		hasLoginSessions	= null;
	private static final ECSessionMananger	instance			= new ECSessionMananger();

	private ECSessionMananger()
	{
		allSessions = new ConcurrentHashMap<String, ECSession>();
		hasLoginSessions = new ConcurrentHashMap<String, ECSession>();
		offlineMessage = new ConcurrentHashMap<String, ConcurrentLinkedQueue<ECPacket>>() ;
	}

	public static ECSessionMananger getInstance()
	{
		return instance;
	}

	public void addTempedSession(String key , ECSession session)
	{
		allSessions.put(key, session);
	}

	public void addAttribute(String sessionId , String key , Object value)
	{
		ECSession session = allSessions.get(sessionId);
		synchronized (session)
		{
			session.setAttribute(key, value);
		}
	}

	public ECSession removeSession(String sessionId) {
		if(allSessions.containsKey(sessionId)) {
			return allSessions.remove(sessionId);
		}
		return null ;
	}
	
	public void removeUserSession(String userCode) {
		if(hasLoginSessions.containsKey(userCode)) {
			hasLoginSessions.remove(userCode);
		}
	}
	
	public ECSession removeTempedSession(String sessionId)
	{
		ECSession s = allSessions.remove(sessionId);
		if (s!=null && s.getConnectedUser() != null)
		{
			hasLoginSessions.remove(s.getConnectedUser());
		}
		return s;
	}

	public ECSession getSession(String key)
	{
		return allSessions.get(key);
	}


	public int getSessionCounts()
	{
		return allSessions.size() + hasLoginSessions.size();
	}

	public ECSession getUserSession(String userCode)
	{
		return hasLoginSessions.get(userCode);
	}
	
	public boolean hasLogin(String userCode) {
		return hasLoginSessions.containsKey(userCode);
	}

	public void reisterValidSession(String key , ECSession session)
	{
		hasLoginSessions.put(key, session);
		sendOfflineMessage(key);
	}
	
	/**
	 * 加入离线消息队列
	 * @param packet
	 */
	public void putOfflineMessage(ECPacket packet){
		System.out.println(packet.getTo() + " 未在线，消息已经加入了离线消息队列，用户上线后将会通知");
		ConcurrentLinkedQueue<ECPacket> messages = offlineMessage.get(packet.getTo());
		if(messages==null) {
			messages = new ConcurrentLinkedQueue<ECPacket>();
			offlineMessage.put(packet.getTo(),messages);
		}
		offlineMessage.get(packet.getTo()).add(packet);
	}
	
	/**
	 * 向用户发送离线消息
	 * @param userCode
	 */
	protected void sendOfflineMessage(final String userCode){
		new Thread(){
			public void run(){
				ConcurrentLinkedQueue<ECPacket> messages = offlineMessage.get(userCode) ;
				ECSession session = getUserSession(userCode);
				if(messages!=null && messages.size()>0) {
					while(!messages.isEmpty()) {
						try {
							ECNioServer.sendPacket(messages.poll(), session.getChannel()) ;
							sleep(5000) ;
						} catch (Exception e) {
						}
					}
					System.out.println(offlineMessage.get(userCode).size());
				}

			}
		}.start();
	}
	
	public int getOnlineUserCount(){
		return hasLoginSessions.size();
	}
	

}
